package com.gugee.tiktok.data.consumer.blogger.listener;

import com.gugee.tiktok.data.common.msg.BloggerMessage;
import com.gugee.tiktok.data.consumer.blogger.config.DataConsumerConfig;
import com.gugee.tiktok.data.consumer.blogger.core.service.BloggerService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.stream.Collectors;


/**
 * 博主获取ai分析年龄、性别主题
 * @author huang xin
 * Date  2020/10/04
 */
@Component
@Slf4j
public class BloggerAiExtMessageListener {

    @Autowired
    BloggerService bloggerService;

    @Autowired
    DataConsumerConfig dataConsumerConfig;

    /**
     * redis服务
     */
    @Autowired
    RedisTemplate<String,String> redisTemplate;

    @KafkaListener(id="${kafka.consumer.group.bloggerAiExt.id}",topics = {"${kafka.topics.bloggerAiExt.online}"},containerFactory = "factoryBloggerAiExtAck")
    public void listen(List<BloggerMessage> messages, Acknowledgment ack) {
        //任务开始时间
        long start = System.currentTimeMillis();
        //任务结果消息
        String error = "success";
        //时间时限
        long acceptMsgTimeSecRange = dataConsumerConfig.getAcceptBloggerAiExtMsgTimeMinRange() * 60 * 1000;
        try {
            long isRange = start - acceptMsgTimeSecRange;
            final int[] noInRangeCount = {0,0};
            List<BloggerMessage> bms = messages.stream().filter(msg->{
                if(msg.getId() == null || msg.getId() <= 0){
                    noInRangeCount[1]++;
                    return false;
                }
                if(msg.getEventTime() < isRange){
                    noInRangeCount[0]++;
                    //超时的要去掉bitmap里的相关记录，保证下次能继续轮到
                    redisTemplate.opsForValue().setBit("blogger_ai_analyze",msg.getId(),false);
                    return false;
                }
                return true;
            }).collect(Collectors.toList());
            if(noInRangeCount[0] > 0){
                log.info("接收到bloggerAiExt消息数量: {},需解析bloggerAiExt消息数量： {},bloggerId无效消息数量：{},不在处理时限内消息数量： {},时限分钟：{}" ,
                        messages.size(),bms.size(), noInRangeCount[1],
                        noInRangeCount[0],dataConsumerConfig.getAcceptBloggerAiExtMsgTimeMinRange());
            }
            //处理
            if(bms.size() > 0){
                bloggerService.bloggerAiExtHandle(bms);
            }
            ack.acknowledge();
        }catch (Exception e){
            log.error("bloggerAiExt 消息处理任务发生异常， 异常 ：{}" ,e.getMessage());
            error = e.getMessage();
        }finally {
            long times = (System.currentTimeMillis() - start) / 1000;
            if(times < 1){
                times = 1;
            }
            log.debug("bloggerAiExt 消息处理，接收到消息数量: {},处理耗时: {}s,结果：{}" ,messages.size(),times,error);
        }
    }
}
